package netty.mq.consumer;

import com.alibaba.fastjson2.JSON;
import com.ruoyi.common.core.dto.ChatMsg;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import netty.session.UserChannelSession;
import netty.websocket.ChatHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.List;

@Slf4j
@Component
public class ChatMessageConsumer {
    @Resource
    private ChatHandler chatHandler;
    @RabbitListener(queues = "${mq.CHAT_ACK_TOPIC}")
    protected void consumer(String message) {
        log.info("netty中的mq接收到ack消息"+message);
        ChatMsg chatMsg = JSON.parseObject(message, ChatMsg.class);
        String senderId = chatMsg.getSenderId();
        String receiverId = chatMsg.getReceiverId();
        //获取发送人id,并获取发送人的channel，提醒发送人已发送完毕
        sendMessageToChannel(senderId,chatMsg);
        sendMessageToChannel(receiverId,chatMsg);
    }

    /**
     * 指定userId找channel发送消息
     */
    private void sendMessageToChannel(String userId,ChatMsg chatMsg){
        List<Channel> multiSession = UserChannelSession.getMultiSession(userId);
        if(!CollectionUtils.isEmpty(multiSession)){
            for (Channel channel : multiSession) {
                Channel existChannel = ChatHandler.clients.find(channel.id());
                if(existChannel!=null){
                    existChannel.eventLoop().execute(()->existChannel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(chatMsg))));
                }
            }
        }
    }
}
